package cn.com.idmy.mq;

import cn.com.idmy.mq.annotation.MqConsumerExt;
import cn.com.idmy.mq.listener.AbstractMsgListener;
import cn.com.idmy.mq.model.enums.MqMsgMode;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.acl.common.AclClientRPCHook;
import org.apache.rocketmq.acl.common.SessionCredentials;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.MessageListener;
import org.apache.rocketmq.client.consumer.rebalance.AllocateMessageQueueAveragely;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.remoting.protocol.heartbeat.MessageModel;
import org.springframework.core.env.Environment;

import java.util.Set;

import static java.text.MessageFormat.format;

@Slf4j
class RocketmqConsumer implements IMqConsumer {
    private final DefaultMQPushConsumer consumer;

    public RocketmqConsumer(MqProps props, String group, MqMsgMode mqMsgMode, String instanceName) {
        AclClientRPCHook rpcHook = new AclClientRPCHook(new SessionCredentials(props.getPassword(), props.getUsername()));
        consumer = new DefaultMQPushConsumer(props.getNamespace(), format("{0}%{1}", props.getNamespace(), group), rpcHook, new AllocateMessageQueueAveragely());
        consumer.setNamesrvAddr(props.getServerAddress());
        consumer.setInstanceName(instanceName);
        consumer.setMaxReconsumeTimes(props.getMaxReconsumeTimes());
        if (mqMsgMode != null) {
            consumer.setMessageModel(MessageModel.valueOf(mqMsgMode.name()));
        }
    }

    public RocketmqConsumer(MqProps props, String group, MqMsgMode mqMsgMode, String instanceName, MqConsumerExt consumerExt, Environment environment) {
        this(props, group, mqMsgMode, instanceName);
        consumer.setConsumeMessageBatchMaxSize(getConsumeMessageBatchMaxSize(consumerExt, environment));
        consumer.setMaxReconsumeTimes(getMaxReconsumeTimes(consumerExt, environment));
    }

    @Override
    public void start() {
        try {
            consumer.start();
        } catch (MQClientException e) {
            log.error("RocketMQ consumer start error:" + e.getErrorMessage(), e);
        }
    }

    @Override
    public void subscribe(String topic, Set<String> tags) {
        try {
            consumer.subscribe(topic, StringUtils.join(tags, "||"));
        } catch (MQClientException e) {
            log.error("RocketMQ consumer subscribe error:" + e.getErrorMessage(), e);
        }
    }

    @Override
    public void setMessageListener(AbstractMsgListener listener) {
        consumer.registerMessageListener((MessageListener) listener);
    }
}
